/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月15日 上午11:17:52
*/
package cn.com.guage.flink.transformation;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月15日 上午11:17:52
*/
public class ConnectTest {
 public static void main(String[] args) throws Exception {
	 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
     env.setParallelism(1);

     DataStreamSource<Integer> integerDataStreamSource = env.fromElements(1, 2, 3);
     DataStreamSource<String> stringDataStreamSource = env.fromElements("Tom", "Jack", "Marry");

     //分别执行map操作
     SingleOutputStreamOperator<Integer> IntDS = integerDataStreamSource.map(new MapFunction<Integer, Integer>() {
         /**
		 * 
		 */
		private static final long serialVersionUID = 8110092523160965085L;

		public Integer map(Integer value) throws Exception {
             return value * value;
         }
     });
     SingleOutputStreamOperator<String> StrDS = stringDataStreamSource.map(new MapFunction<String, String>() {

		@Override
		public String map(String value) throws Exception {
			return "hello: " + value;
		}
    	 
	});

     //connect合流操作
     /**
      *  union多流合并 connect双流合并
      * (1)、只能用于连接两个DataStream流，不能用于DataSet;
		(2)、连接的两个数据流数据类型可以不同
		(3)、连接的结果为一个ConnectedStreams流
		(4)、连接后两个流可以使用不同的处理方法，两个流可以共享状态；
		(5)、连接后可以使用CoMapFunction或CoFlatMapFunction等方法进行处理
      */
     ConnectedStreams<Integer, String> connectDS = IntDS.connect(StrDS);
     SingleOutputStreamOperator<Object> res = connectDS.map(new CoMapFunction<Integer, String, Object>() {
         /**
		 * 
		 */
		private static final long serialVersionUID = 1L;

		//分别对两条流进行操作
         public Object map1(Integer value) throws Exception {
           
             return "二次平方后的结果为: "+ value*value;
         }

         @Override
         public Object map2(String value) throws Exception {
        	  return value;
         }
     });
     res.print();
     env.execute();
}
}
